package repeater

import (
    "../common"
    "fmt"
    "github.com/influxdata/influxdb/client/v2"
    "strings"
    "time"
)

type InfluxWriter struct {
    dbname        string
    user          string
    passwd        string
    addr          string
    workingOffset int64
    filter        *Filter
    rc            *RepeaterController
    cli           client.Client
}

func NewInfluxWriter(d map[string]string) (*InfluxWriter, error) {
    dbname := d["dbname"]
    user := d["user"]
    passwd := d["passwd"]
    addr := d["addr"]
    if dbname == "" || user == "" || passwd == "" || addr == "" {
        return nil, fmt.Errorf("missing config")
    }
    var filter *Filter
    filterPath := d["filter"]
    if filterPath != "" {
        f, err := LoadFilter(filterPath)
        if err != nil {
            return nil, err
        }
        filter = f
    }
    return &InfluxWriter{
        dbname: dbname,
        user:   user,
        passwd: passwd,
        addr:   addr,
        filter: filter,
    }, nil
}

func (ww *InfluxWriter) Init() (err error) {
    ww.cli, err = client.NewHTTPClient(client.HTTPConfig{
        Addr:     ww.addr,
        Username: ww.user,
        Password: ww.passwd,
    })
    return
}

func (ww *InfluxWriter) Start() {
}

func (ww *InfluxWriter) Stop() (err error) {
    return ww.cli.Close()
}
func (ww *InfluxWriter) Name() string {
    return "influx" + ww.dbname
}
func (ww *InfluxWriter) Handle(dps *DataPointPack) error {
    bp, err := client.NewBatchPoints(client.BatchPointsConfig{
        Database:  ww.dbname,
        Precision: "m",
    })
    if err != nil {
        return err
    }
    for _, dp := range dps.Datapoints {
        ret := common.Extract(dp.Name)
        ip := ret["ip"]
        head := ret["head"]
        sub := ret["sub"]
        path := ret["path"]
        if ip == "" || head == "" || sub == "" {
            continue
        }
        if ww.filter != nil && !ww.filter.pass(dp.Name) {
            continue
        }
        tags := make(map[string]string)
        tags["ip"] = ip
        if path != "" {
            if strings.Contains(head, "disk") {
                if p, err := common.Decode_base64(path); err == nil {
                    path = p
                }
            }
            tags["path"] = path
        }
        fields := map[string]interface{}{sub: dp.Val}
        pt, err := client.NewPoint(head, tags, fields, time.Unix(int64(dp.Time), 0))
        if err != nil {
            log.Warningf("ilegal datapoint %v", dp)
            continue
        }
        bp.AddPoint(pt)
    }
    //log.Infof("dp size %d",len(bp.Points()))
    if err := ww.cli.Write(bp); err != nil && !IsBeyondRetention(err){
        return err
    }
    ww.workingOffset = dps.Offset()
    return nil
}

func IsBeyondRetention(err error)bool{
    s:=err.Error()
    return strings.Contains(s,"partial write: points beyond retention policy")
}

func (ww *InfluxWriter) SetController(rc *RepeaterController) error {
    ww.rc = rc
    return nil
}

func (ww *InfluxWriter) Commit() (int64, error) {
    return ww.workingOffset, nil
}
